package com.xiaobai.flink;

import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.DebeziumSourceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FlinkCDC {
    public static void main(String[] args) throws Exception {
        // 1.获取Flink 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 2.通过FlinkCDC构建SourceFunction
        DebeziumSourceFunction<DataChangeInfo> sourceFunction = MySqlSource.<DataChangeInfo>builder().hostname("127.0.0.1")
                .port(3306).username("root").password("123456").serverTimeZone("GMT+8").databaseList("flink")
                .tableList("user").deserializer(new MysqlDeserialization())
                 .startupOptions(StartupOptions.latest())
                .build();

        DataStreamSource<DataChangeInfo> dataStreamSource = env.addSource(sourceFunction);

        // 3.数据打印
        dataStreamSource.print();

        dataStreamSource.addSink(new DataChangeSink());

        // 4.启动任务
        env.execute("FlinkCDC");
    }
}
